from multiprocessing import Pool, Manager
from lxml import etree
from queue import Queue
from requests.auth import HTTPProxyAuth
from tool import *
from cjjtool import jx_phtml2
from sql import sqlseverDB
import requests, math
import traceback,logging

ms = sqlseverDB('{SQL Server}', SQL_SERVER, 'SuMaiTongPol', SQL_USER, SQL_PWD)

mutil_count=5

logger = logging.getLogger()
logger.setLevel(logging.INFO)  # 设置日志级别
log_path=f'日志文件/亚马逊CJPro.log'
file_handler = logging.FileHandler(log_path, mode='a', encoding='utf-8')

# 创建控制台处理器 (Handler)
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)

# 设置日志格式
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)

# 将处理器添加到 logger 对象
logger.addHandler(file_handler)
logger.addHandler(console_handler)

def process_shop(bid,kitems,queue,task_count):
    logging.info(f"进程启动: bid={bid}, 任务数={len(kitems)}")
    try:
        res= asyncio.run(reqPro(bid,kitems,queue,task_count))
        logging.info(res)

    except Exception as e:
        logging.error(f'bid:{bid}, 出现错误: {e}')
        traceback.print_exc()
        time.sleep(3)

async def reqPro(bid,pitems,ycj_queue,task_count):

    browser=await openSunBrowser(bid)
    page=await browser.newPage()

    for pitem in pitems:
        pid, asin, purl,mid = pitem

        x_ind=ycj_queue.qsize()+1
        try:
            await page.goto(purl,{'timeout':0})
            await asyncio.sleep(3)
        except Exception as e:
            logging.info(f'第{x_ind}个,第 {x_ind} 个ASIN:{asin},,请求出错 => {e}')
            ycj_queue.put(1)
            latest_count=task_count-ycj_queue.qsize()
            continue
        
        elSeller=None
        while True:
            cot=await page.content()
            el_tit=await page.xpath('//span[@id="productTitle"]')
            if el_tit or len(cot)>20000:
                elSeller=True
                break
            
            cur_tit= await page.title()
            if cur_tit=='Sorry! Something went wrong!':
                aff_503=ms.ExecNoQuery('update [AmazonDPAsin] set state=503 where [ASIN]=?', (asin,))
                logging.info(f'第{x_ind}个,第 {x_ind} 个ASIN:{asin},页面503,标记状态503:{aff_503}')
                break
            if 'Page Not Found' in cur_tit:
                aff_404=ms.ExecNoQuery('update [AmazonDPAsin] set state=404 where [ASIN]=?', (asin,))
                logging.info(f'第{x_ind}个,第 {x_ind} 个ASIN:{asin},页面404,标记状态2:{aff_404}')
            cot= await page.content()
            if len(cot)<20000:
                aff_lower=ms.ExecNoQuery('update [AmazonDPAsin] set state=101 where [ASIN]=?', (asin,))
                logging.info(f'第{x_ind}个,第 {x_ind} 个ASIN:{asin},页面内容过少,标记状态101:{aff_lower}')
                break
        
        if not elSeller:
            ycj_queue.put(1)
            latest_count=task_count-ycj_queue.qsize()
            logging.info(f'剩余任务数:{latest_count}/{task_count}\n')
            continue


        try:
            pitem=jx_phtml2(asin,mid,cot)
            rs=ms.ExecQuery('select count(*) from proshopeeph where itemid=?',(asin,))
            if rs[0][0]==0:
                zdstr=','.join(pitem.keys())
                params=list(pitem.values())
                wstr=','.join(['?' for _ in params])
                sqlstr=f'insert into proshopeeph({zdstr}) values({wstr})'
                aff= ms.ExecNoQuery(sqlstr,params)
            else:
                aff=f'产品已存在'
            aff_cg=ms.ExecNoQuery('update [AmazonDPAsin] set state=1 where [ASIN]=?', (asin,))
            ycj_queue.put(1)
            latest_count=task_count-ycj_queue.qsize()
            logging.info(f'第 {x_ind} 个ASIN:{asin},采集完毕,存入状态:{aff},更改ASIN状态:{aff_cg},剩余任务数:{latest_count}/{task_count}\n')
        except Exception as e:
            ycj_queue.put(1)
            latest_count=task_count-ycj_queue.qsize()
            logging.info(f'第 {x_ind} 个ASIN:{asin},解析错误=>{e},剩余任务数:{latest_count}/{task_count}\n')
            traceback.print_exc()

    await page.close()
def main():
    
    logging.info(f'第四步采集店铺产品....')
    rs = ms.ExecQuery('select [id], [ASIN], [Purl],[mid] from [AmazonDPAsin] where State=0')

    taskCount = len(rs)

    logging.info(f'本次采集,待采集ASIN总数: {taskCount}')

    #创建采集环境
    while True:
        all_bids=get_browser_environment()
        if len(all_bids)>=mutil_count:
            bids=all_bids[:mutil_count]
            break
        else:
            logging.info(mutil_count-len(all_bids))
            for _ in range(mutil_count-len(all_bids)):
                res=create_browser_environment_no_proxy(f'亚马逊采集{get_random_code()}')
                time.sleep(1)
    
    logging.info(f'成功创建:{len(bids)}个采集环境')

    manager = Manager()
    queue = manager.Queue()

    qu_jian=math.ceil(taskCount/mutil_count)
    with Pool(processes=mutil_count) as pool:  # 根据实际情况设置并发进程数量
        results = []
        for j in range(mutil_count):
            pitems=rs[j*qu_jian:(j+1)*qu_jian]
            result = pool.apply_async(
                process_shop, args=(bids[j],pitems,queue,taskCount))
            results.append(result)

        # 监控任务执行情况
        pool.close()
        for result in results:
            result.wait()  # 等待每个进程完成，无超时限制

        # 所有任务执行完成后继续
        pool.join()

if __name__ == '__main__':
    main()
